package com.wingsboat.data.transmission.consumer;

import com.wingsboat.data.transmission.common.constant.ConsumerTopicConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * @Author: wangzhong
 * @Date: 2023/11/7 15:11
 * @Description:
 **/

@Component
public class MessageKafkaConsumer {
    private static final Logger log = LoggerFactory.getLogger(MessageKafkaConsumer.class);


    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_C1})
    public void get1CParams(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[1C检测数据]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_C6})
    public void get1CDefect(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[缺陷报表]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_DEDUCTION})
    public void get1CKmDeduction(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[每公里扣分数统计]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_CDI})
    public void get1CCDI(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[单元CDI数据]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_C1_WAVE})
    public void get1CWave(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[1C检测原始数据wave文件]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_C1_CHECK})
    public void get1CParamsCheck(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[1C检测数据对账]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_C6_CHECK})
    public void get1CDefectCheck(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[缺陷报表对账]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_DEDUCTION_CHECK})
    public void get1CKmDeductionCheck(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[每公里扣分数统计对账]----------record.key="+record.value().toString());
    }

    @Async("wingsBoatAsync")
    @KafkaListener(topics = {
            ConsumerTopicConstant.WINGSBOAT_MSG_CDI})
    public void get1CCDICheck(ConsumerRecord<?, ?> record) {
        log.info("开始预处理[单元CDI数据对账]----------record.key="+record.value().toString());
    }
}
